package com.cxb.springboot.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.cxb.springboot.common.ZookeeperClient;
import com.cxb.springboot.domin.AntiRequest;
import com.cxb.springboot.listener.MyPathChildrenCacheListener;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.atomic.DistributedAtomicLong;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.queue.DistributedQueue;
import org.apache.curator.framework.recipes.queue.QueueBuilder;
import org.apache.curator.framework.recipes.queue.QueueConsumer;
import org.apache.curator.framework.recipes.queue.QueueSerializer;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.retry.RetryNTimes;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;

/**
 * @Classname ZkConfiguration
 * @Description TODO
 * @Date 2023/3/28 21:57
 * @Created by Administrator
 */
@Configuration
@Slf4j
public class ZkConfiguration {

    @Resource
    private WrapperZk wrapperZk;

    /**
     * 其中RetryPolicy为重试策略，第一个参数为baseSleepTimeMs初始的sleep时间，
     * 用于计算之后的每次重试的sleep时间。第二个参数为maxRetries，最大重试次数。
     *
     * @return
     */
    /*@Bean(initMethod = "start")
    public CuratorFramework curatorFramework() {
        RetryPolicy retrYPolicy = new ExponentialBackoffRetry(wrapperZk.getBaseSleepTimeMs(),
                wrapperZk.getRetryCount());
        CuratorFramework client = CuratorFrameworkFactory
                .newClient(wrapperZk.getConnectString(), retrYPolicy);
        log.info("zk curator初始化完成...");
        return client;
    }*/

    @Resource
    private ZookeeperProperties zookeeperProperties;

    private static final String rootPath = "/root";

    private static final String dataPath = "/root/data";

    private static final String nodePath = "/root/data/stock";

    @SneakyThrows
    @Bean
    public CuratorFramework curatorFrameworkClient() {
        //重试策略,ExponentialBackoffRetry(1000,3)这里表示等待1s重试，最大重试次数为3次
        RetryPolicy policy = new ExponentialBackoffRetry(zookeeperProperties.getSleepTime(),
                zookeeperProperties.getMaxRetries());
        //构建CuratorFramework实例
        CuratorFramework curatorFrameworkClient = CuratorFrameworkFactory
                .builder()
                .connectString(zookeeperProperties.getServer())
                .sessionTimeoutMs(zookeeperProperties.getSessionTimeout())
                .connectionTimeoutMs(zookeeperProperties.getConnectionTimeout())
                .retryPolicy(policy)
                .build();
        // 启动实例
        curatorFrameworkClient.start();

        // 该监听方式直接替换了下面的三种过时的监听方式
        CuratorCache curatorCache = CuratorCache.build(curatorFrameworkClient, dataPath);
        CuratorCacheListener listener = CuratorCacheListener
                // 1、监听指定的节点
                // .builder().forNodeCache(new MyNodeCacheListener())
                // 2、监听指定节点及其子节点的数据变化
                .builder().forPathChildrenCache(dataPath, curatorFrameworkClient, new MyPathChildrenCacheListener())
                // 3、监听指定节点及其子节点孙子节点的数据变化
                // .builder().forTreeCache(curatorFrameworkClient, new MyTreeCacheListener())
                .build();
        curatorCache.listenable().addListener(listener);
        curatorCache.start();

        // 监听指定节点的变化
        /*NodeCache nodeCache = new NodeCache(curatorFrameworkClient, nodePath);
        nodeCache.getListenable().addListener(new MyNodeCacheListener());
        nodeCache.start();*/

        // 监听节点及子节点的变化
        /*PathChildrenCache pathChildrenCache = new PathChildrenCache(curatorFrameworkClient, dataPath,
                true);
        pathChildrenCache.getListenable().addListener(new MyPathChildrenCacheListener());
        pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);*/

        // 监听指定节点和子节点及其子孙节点
        /*TreeCache treeCache = new TreeCache(curatorFrameworkClient, rootPath);
        treeCache.getListenable().addListener(new MyTreeCacheListener());
        treeCache.start();*/

        /*CuratorCache curatorCache = CuratorCache.builder(curatorFrameworkClient, rootPath).build();
        curatorCache.listenable().addListener(new MyCuratorCacheListener());
        curatorCache.start();*/

        return curatorFrameworkClient;
    }

    /**
     * 采用这种方式注册bean可以比较优雅的关闭连接
     */
    @Bean(destroyMethod = "destroy")
    public ZookeeperClient zookeeperClient(CuratorFramework curatorFrameworkClient) {
        return new ZookeeperClient(curatorFrameworkClient);
    }

    @Bean("distributedQueue")
    public DistributedQueue createQueue(CuratorFramework curatorClient) {
        DistributedQueue<AntiRequest> distributedQueue = null;
        try {
            QueueConsumer<AntiRequest> consumer = createQueueConsumer();
            distributedQueue = QueueBuilder.builder(curatorClient, consumer, createQueueSerializer(),
                    getQueuePath()).buildQueue();
            distributedQueue.start();
        } catch (Exception ex) {
            log.error("创建队列异常", ex);
        }
        return distributedQueue;
    }

    /**
     * @return
     */
    private String getQueuePath() {
        return "/HRHK_ZT/ST";
    }

    /**
     * @return
     */
    private QueueSerializer<AntiRequest> createQueueSerializer() {
        return new QueueSerializer<AntiRequest>() {
            @Override
            public byte[] serialize(AntiRequest item) {
                return JSON.toJSONString(item).getBytes(StandardCharsets.UTF_8);
            }

            @Override
            public AntiRequest deserialize(byte[] bytes) {
                return JSONObject.parseObject(new String(bytes), AntiRequest.class);
            }

        };
    }

    /**
     * @return
     */
    private QueueConsumer<AntiRequest> createQueueConsumer() {
        return new QueueConsumer<AntiRequest>() {
            @Override
            public void stateChanged(CuratorFramework client, ConnectionState newState) {
                log.info("状态改变，当前状态：{}", newState.name());
            }

            @Override
            public void consumeMessage(AntiRequest callbackRequest) {
                log.info("消费的消息：{}", callbackRequest);
            }
        };
    }

    /**
     * 分布式ID
     *
     * @return
     */
    @Bean
    public DistributedAtomicLong distributedIdGenerator() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework curatorFramework = CuratorFrameworkFactory.newClient(zookeeperProperties.getServer(), retryPolicy);
        curatorFramework.start();

        DistributedAtomicLong distributedIdGenerator = new DistributedAtomicLong(curatorFramework,
                "/id-generator", new RetryNTimes(3, 1000));
        return distributedIdGenerator;
    }

}

